package cn.musijas.service.api;

import cn.musi.constant.RedisKeysConstant;
import cn.musi.order.api.IOrderService;
import cn.musijas.service.entity.Product;
import cn.musijas.service.entity.ProductRecord;
import cn.musijas.service.mapping.ProductMapping;
import cn.musijas.service.mapping.ProductRecordMapping;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.persistence.criteria.Order;

@Transactional(rollbackFor = Exception.class)
@Service
public class OrderServiceImpl implements IOrderService {
    private Logger  logger= LoggerFactory.getLogger(OrderServiceImpl.class);

    @Value("${kafka.topic}")
    private String kafkaTopic ;
    @Autowired
    ProductMapping  productService;
    @Autowired
    ProductRecordMapping  productRecordService;
    @Autowired
    private RedisTemplate<String,String> redisTemplate ;
    @Autowired
    private KafkaProducer kafkaProducer ;

    @Override
    public int createWrongOrder(int sid) throws Exception {
         //校验库存
        Product stock = checkStock(sid);
        //扣库存
        saleStock(stock);
        //创建订单
        int id = createOrder(stock);
        return id;
    }

    @Override
    public int createOptimisticOrder(int sid) throws Exception {
        Product product=checkStock(sid);
        saleStockOptimistic(product);
        return   createOrder(product);
    }

    @Override
    public int createOptimisticOrderUseRedis(int sid) throws Exception {
        Product  product=checkStockByRedis(sid);

        //乐观锁更新库存 以及更新 Redis
        saleStockOptimisticByRedis(product);
        //创建订单
        int id = createOrder(product);
        return id;
    }


    @Override
    public void createOptimisticOrderUseRedisAndKafka(int sid) throws Exception {
        //检验库存，从 Redis 获取
        Product stock = checkStockByRedis(sid);
        //利用 Kafka 创建订单
        kafkaProducer.send(new ProducerRecord<String,Object>(kafkaTopic,stock)) ;
        logger.info("send Kafka success");
    }

    private Product checkStock(int sid) {
        Product stock = productService.selectByPrimaryKey(sid);
        if (stock.getSale().equals(stock.getCount())) {
            throw new RuntimeException("库存不足");
        }
        return stock;
    }

    private void saleStockOptimistic(Product stock) {
        int count = productService.updateByOptimistic(stock);
        if (count == 0){
            throw new RuntimeException("并发更新库存失败") ;
        }
    }

    private int createOrder(Product stock) {
        ProductRecord order = new ProductRecord();
        order.setSid(stock.getId());
        order.setName(stock.getName());
        int id = productRecordService.insertSelective(order);
        return id;
    }

    private Product checkStockByRedis(int sid) throws Exception {
        Integer count = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_COUNT + sid));
        Integer sale = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_SALE + sid));
        if (count.equals(sale)){
            throw new RuntimeException("库存不足 Redis currentCount=" + sale);
        }
        Integer version = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_VERSION + sid));
        Product stock = new Product() ;
        stock.setId(sid);
        stock.setCount(count);
        stock.setSale(sale);
        stock.setVersion(version);

        return stock;
    }

    /**
     * 乐观锁更新数据库 还要更新 Redis
     * @param stock
     */
    private void saleStockOptimisticByRedis(Product stock) {
        int count = productService.updateByOptimistic(stock);
        if (count == 0){
            throw new RuntimeException("并发更新库存失败") ;
        }
        //自增
        redisTemplate.opsForValue().increment(RedisKeysConstant.STOCK_SALE + stock.getId(),1) ;
        redisTemplate.opsForValue().increment(RedisKeysConstant.STOCK_VERSION + stock.getId(),1) ;
    }

    private int saleStock(Product stock) {
        stock.setSale(stock.getSale() + 1);
        return productService.updateByPrimaryKeySelective(stock);
    }

}
